package com.atguigu.edu.realtime.app.dws.app;

import com.atguigu.edu.realtime.app.dws.function.KeywordUDTF;
import com.atguigu.edu.realtime.common.base.BaseSqlApp;
import com.atguigu.edu.realtime.common.bean.KeywordBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.JdbcUtil;
import com.atguigu.edu.realtime.common.util.SqlUtil;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(
                11103,
                4,
                "dws_traffic_source_keyword_page_view_window"
                , Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);
        //从kafka读取页面浏览日志
        tableEnv.executeSql("create table page_log(\n" +
                "`common` map<string, string>,\n" +
                "`page` map<string, string>,\n" +
                "`ts` bigint,\n" +
                "row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000, 'yyyy-MM-dd HH:mm:ss')),\n" +
                "WATERMARK FOR row_time AS row_time - INTERVAL '3' SECOND\n" +
                ")" + SqlUtil.getKafkaDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, "dws_traffic_source_keyword_page_view_window"));
//        String sql = "select * from page_log";
//        tableEnv.executeSql(sql).print();
        //从表中过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
                "page['item'] full_word,\n" +
                "row_time\n" +
                "from page_log\n" +
                "where page['item'] is not null\n" +
                "and page['item_type'] = 'keyword'");
        tableEnv.createTemporaryView("search_table",searchTable);

        //使用自定义的udtf函数对搜索的内容分词
        Table splitTable = tableEnv.sqlQuery("select\n" +
                "keyword,\n" +
                "row_time \n" +
                "from search_table,\n" +
                "lateral table(ik_analyze(full_word))\n" +
                "as t(keyword)");
        tableEnv.createTemporaryView("split_table", splitTable);


        //分组开窗聚合计算
        Table KeywordBeanSearch = tableEnv.sqlQuery("SELECT\n" +
                "    DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') stt,\n" +
                        "    DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') edt,\n" +
                        "    DATE_FORMAT(window_start, 'yyyy-MM-dd') cur_date, \n" +
                        "    keyword,\n" +
                        "    count(*) keyword_count\n" +
                        "  FROM TABLE(\n" +
                        "    TUMBLE(TABLE split_table, DESCRIPTOR(row_time), INTERVAL '10' second))\n" +
                        "  GROUP BY keyword,window_start, window_end"
        );
        tableEnv.createTemporaryView("keyword_bean_search", KeywordBeanSearch);
        //tableEnv.sqlQuery("select * from keyword_bean_search").execute().print();

//        tableEnv.executeSql("create table dws_traffic_source_keyword_page_view_window(\n" +
//                "     stt string,\n" +
//                "     edt string,\n" +
//                "     cur_date string,\n" +
//                "     keyword string,\n" +
//                "     keyword_count BIGINT NOT NULL\t\n" +
//                ") " + SqlUtil.getDorisDDL("dws_traffic_source_keyword_page_view_window")).print();
        KeywordBeanSearch.executeInsert("edu_realtime.dws_traffic_source_keyword_page_view_window");
    }
}
